package org.niit.service

import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.AdClickData
import org.niit.dao.UserAdDao

import java.text.SimpleDateFormat
import java.util.Date

/**
 * Date:2025/6/3
 * Author：Ys
 * Description:
 */
class UserBlackListService {




  //声明一个总方法，分别调用不同的处理数据的逻辑
  def dataAnalysis(data:DStream[AdClickData]): Unit ={

    //1. 过滤出没有被拉黑的数据，并统计点击某广告的次数
    val ds: DStream[((String, String, String), Int)] = filterData(data)
    //2.检查用户点击广告次数 是否拉黑。
    checkUserAdCount(ds)

  }
  def filterData(data: DStream[AdClickData]): DStream[((String,String,String),Int)] = {
    //将DStream 转成 RDD 再进行过滤和计算  ，回顾下RDD的操作
    data.transform(rdd=>{
      //1.过滤出没有被拉黑的用户
      val filterRDD  = rdd.filter(line=>{
        //需要读取数据库中黑名单表，将不在黑名单中的用户过滤出来
          val userAdDao = new UserAdDao
          !userAdDao.selectBlackUserById(line.user)
      })
      //filterRDD 是装有没有被拉黑的用户
      //根据没有被拉黑的用户 去统计 某时间 某用户 点击 某广告总次数
      // 单词统计Key word   value : 1     Key：用户 时间 广告           value: 1
      val redRDD: RDD[((String, String, String), Int)] = filterRDD.map(line => { // line== >AdClickData
        val user = line.user //获得用户
        val ad = line.ad
        //将样例类中时间戳 转换成 年-月-日
        val sdf = new SimpleDateFormat("yyyy-MM-dd")
        val day = sdf.format(new Date(line.ts.toLong)) //Date 是 java.util.Date
        ((day, user, ad), 1)
      })//已经获得 某用户 在某时间 点击了 某广告   1
      redRDD.reduceByKey( _ + _)
    })


  }

  /*
     如果用户 点击 某广告 一次性 大于 100次 拉入黑名单 --》 存入黑名单表
     如果用户 累计 点击 某广告   大于 100次 拉入黑名单 --》 存入黑名单表
        如果用户 点击某个 广告次数  小于 100次  存入用户每日点击广告次数表中
   */
  def checkUserAdCount(ds: DStream[((String, String, String), Int)]): Unit = {

    ds.foreachRDD(rdd=>{
      rdd.foreach{
        case ( (day,user,ad),count )=>{
          //1.如果用户 点击 某广告 一次性 大于 100次 拉入黑名单 --》 存入黑名单表
          val userAdDao = new UserAdDao
          if(count > 100){
            //存入黑名单表
            userAdDao.insertBlackUser(user)
          }else{
            // 存入用户每日点击广告次数表中
            userAdDao.insertUserAdCount(day,user,ad,count)
          }
          //判断更新后的点击数据是否超过阈值，如果超过阈值，则拉入黑名单  数据库的操作
          val bool = userAdDao.checkUserAdCount(day,user,ad,100)
          if(bool){
            userAdDao.insertBlackUser(user)
          }
        }

      }
    })

  }
}
